/**
 * 
 */
package com.skivingcloud.file.store;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;

import com.skivingcloud.file.FilePathUtil;
import com.skivingcloud.file.entity.SysFile;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;


/**
 * 文件存储方案的实现，hdfs存储
 * @author hushouquan
 *
 */
@Service("hdfsFileStorageDao")
public class HdfsFileStorageDaoImpl implements FileStorageDao {

	
	@Value("${hdfs.fs.defaultFS:hdfs://mycluster}")
	private String defaultFS;
	@Value("${hdfs.dfs.blocksize:3}")
	private int blocksize;
	@Value("${hdfs.dfs.replication:3}")
	private int replication;
	@Value("${hdfs.dfs.nameservice:mycluster}")
	private String nameservice;
	@Value("${hdfs.dfs.namenodes.id:nn1,nn2}")
	private String namenodes;
	@Value("${hdfs.dfs.namenodes.url:127.0.0.1:8020,127.0.0.1:8020}")
	private String namenodesUrl;
	@Value("${hdfs.HADOOP_USER_NAME:sysadm}")
	private String userName;
	@Value("${hdfs.dfs.file.root:/}")
	private String fileRoot;
	
	private static final int bufferSize = 1024 * 1024 * 64;
	
	/**
	 * 获取HDFS文件系统对象
	 * @return
	 * @throws IOException
	 * @throws URISyntaxException
	 */
	public FileSystem getFileSystem() throws IOException, URISyntaxException{
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS",defaultFS);
        conf.setInt("dfs.blocksize",blocksize );
        conf.setInt("dfs.replication", replication);
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");//hdfs文件操作类设置
        //以下几项是线上hadoop HA配置nameservice，如果是测试环境单节点的话，直接defaultFS几个属性都只指定具体namenode的ip即可，注意这里的nameservice1是命名空间的名字，
        //这个可以从集群配置文件hdfs-site.xml中找到,以下参数都可以在配置文件中找到（hdfs-site.xml或core-site.xml）
        conf.set("dfs.nameservices",nameservice);
        conf.set("dfs.ha.namenodes." + nameservice, namenodes);
        //dfs.namenode.rpc-address.[nameservice ID].[name node ID]每一个namenode监听的标准RPC地址
        String[] nnArr = namenodes.split(",");
        String[] nnUrlArr = namenodesUrl.split(",");
        for(int i = 0; i < nnArr.length; i++) {
        	String nn = nnArr[i];
        	conf.set("dfs.namenode.rpc-address." + nameservice + "." + nnArr[i], nnUrlArr[i]);//原生默认端口应该是9000，我这里cdh应该是自动给了8020
        }
    	//参数dfs.client.failover.proxy.provider.[nameservice ID]，该参数定义HDFS客户端用来和活动的namenode联系的java类。配置的java类是用来给HDFS客户端判断哪个namenode节点是活动的，当前是哪个namenode处理客户端的请求
    	conf.set("dfs.client.failover.proxy.provider." + nameservice, "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
    	// 通过设置当前JVM系统环境变量来指明客户身份
    	System.setProperty("HADOOP_USER_NAME",userName);
    	
    	URI uri = new URI(defaultFS);
        FileSystem fs = FileSystem.get(uri, conf);
        return fs;
	}
	
	/**
	 * 保存文件的方法
	 * @param fileManagement
	 * @param fileInputStream 
	 * @throws Exception 
	 */
	@Override
	public void saveFile(SysFile fileManagement, InputStream fileInputStream) throws Exception {
		String basePath = FilePathUtil.getHdfsFileFolder(fileRoot);
		FileSystem fs = null;
		FSDataOutputStream outputStream = null;
		String realFilePath = basePath + fileManagement.getId() + "@" + fileManagement.getFileName();
		try {
			Path newPath = new Path(realFilePath);
			fs = getFileSystem();
			outputStream = fs.create(newPath);
			// 打开一个输出流
			byte[] buffer = new byte[fileInputStream.available()];
			fileInputStream.read(buffer);
			outputStream.write(buffer);
			fileManagement.setHdfsPath(realFilePath);
		} catch (Exception e) {
			throw e;
		} finally {
			if(outputStream != null) {
				outputStream.close();
			}
			if(fs != null) {
				fs.close();
			}
			if(fileInputStream != null) {
				fileInputStream.close();
			}
		}
	}

	/**
	 * 读取文件的方法
	 * @param fileManagement
	 * @return
	 * @throws Exception 
	 */
	@Override
	public InputStream getFile(SysFile fileManagement) throws Exception {
		FileSystem fs = null;
		try {
			// 目标路径
			Path srcPath = new Path(fileManagement.getHdfsPath());
			fs = getFileSystem();
			return fs.open(srcPath);
		} catch (Exception e) {
			throw e;
		} finally {
			if(fs != null) {
				fs.close();
			}
		}
	}

	/**
	 * 作废或删除文件，实际上是把文件移动到“作废”或“已删除”文件夹下
	 * @param fileManagement
	 * @param fileState
	 * @throws Exception 
	 */
	@Override
	public void moveFilePath(SysFile fileManagement, String fileState) throws Exception {
		String fileRootNew = fileRoot + (fileState.equals("2") ? "作废" : "已删除" + "/");
		FileSystem fs = null;
		// 原始文件路径
        Path oldPath = new Path(fileManagement.getHdfsPath());
        // 目标路径
        Path newPath = new Path(fileManagement.getHdfsPath().replaceFirst(fileRoot, fileRootNew));

        FSDataInputStream inputStream = null;
        FSDataOutputStream outputStream = null;
        
        try {
            inputStream = fs.open(oldPath);
            outputStream = fs.create(newPath);

            IOUtils.copyBytes(inputStream, outputStream, bufferSize, false);
            fs.delete(oldPath, true);
            fileManagement.setHdfsPath(fileManagement.getHdfsPath().replaceFirst(fileRoot, fileRootNew));
        } finally {
            inputStream.close();
            outputStream.close();
            fs.close();
        }
	}
	
	

}
